package thss.mro.bolt;

import java.util.Map;

import thss.mro.entity.Packet;
import thss.mro.entity.template.ProtocolHeadTemplate;
import thss.mro.manager.ProtocolManager;
import thss.mro.manager.TemplateManager;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

/**
 * 解析头部的bolt
 */
public class PacketHeadParse extends BaseRichBolt {

	private static final long serialVersionUID = -900109618474202608L;
	OutputCollector _collector;
	
	@Override
	public void execute(Tuple tuple) {
		//从流中接收数据:headTplId-头部模板ID；data-数据包
		String headTplId = tuple.getString(0);
		byte[] data = (byte[])tuple.get("data");
		//获得头部模板并解析头部
		ProtocolHeadTemplate headTpl = (ProtocolHeadTemplate)TemplateManager.getTemplate(headTplId);
		Packet headPacket = ProtocolManager.parse(headTpl, data);
			//TODO:打印解析头部
		//根据头部模板得到正文模板ID及得到正文数据
		String bodyTplId = TemplateManager.getBodyTemplate(headTpl,headPacket);
		byte[] bodyData =  new byte[data.length-headTpl.getLength()];
		System.arraycopy(data, headTpl.getLength(), bodyData, 0, bodyData.length);
		//输出
		_collector.emit(tuple, new Values(bodyTplId,bodyData));//向下游发送协议类型与数据包
		_collector.ack(tuple);
	}

	@SuppressWarnings("unchecked")
	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		_collector = collector;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("bodyTplId","bodyData"));
	}

}
